package com.hk.rabbitmq.four;

import com.hk.rabbitmq.utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @author: dpb
 * @date: 2022/3/18
 *
 *  发布确认：
 *   1、单个确认发布
 *
 *   2、批量确认发布
 *
 *   3、异步确认发布
 */
public class ConfirmMessage {

    //批量发送消息的个数
    public static final int message_count = 1000;


    public static void main(String[] args) throws Exception {

//        1、单个确认发布
//        publishMessageIndividually(); //发布1000个单独确认消息，耗时10087ms
//        2、批量确认发布
//        publicMessageBatch();  //发布1000个批量消息，耗时206ms
//        3、异步确认发布
        publishMessageAsync();//发布1000个批量消息，耗时63ms


    }

    public static void publishMessageIndividually() throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString();
        //创建队列
        channel.queueDeclare(queueName, false, false, false, null);

        //开启确认发布
        channel.confirmSelect();
        long begin = System.currentTimeMillis();
        for (int i = 0; i < message_count; i++) {
            String message = String.valueOf(i);
            //创建消息
            channel.basicPublish("", queueName, null, message.getBytes());
            //单个消息马上确认
            boolean flag = channel.waitForConfirms();
            if (flag){
                System.out.println("消息发送成功！");
            }
        }
        long end = System.currentTimeMillis();
        System.out.println("发布" +message_count+"个单独确认消息，耗时" + (end - begin) + "ms");
    }

    public static void publicMessageBatch() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,false,false,false,null);

        channel.confirmSelect();
        int batchSize = 100;
        int outStandingMessageCount = 0;
        long begin = System.currentTimeMillis();
        for (int i = 0; i < message_count; i++) {
            String message = String.valueOf(i);
            channel.basicPublish("", queueName, null, message.getBytes());
            //每100次就确认
            if ((i + 1) % batchSize == 0) {
                //确认
                channel.waitForConfirms();
            }
        }
        long end = System.currentTimeMillis();
        System.out.println("发布" +message_count+"个批量消息，耗时" + (end - begin) + "ms");
    }

    //异常发布确认
    public static void publishMessageAsync() throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,false,false,false,null);

        //开启发布确认
        channel.confirmSelect();

        /**
         * 线程安全有序的一个哈希表，适用于高并发的情况下
         * 1.将序列号与消息进行关联
         * 2.批量删除条目，只要有序列号
         * 3.支持并发访问
         */
        ConcurrentSkipListMap<Long, String> outStandingConfirms = new ConcurrentSkipListMap<>();

        //消息确认成功 回调函数
        ConfirmCallback ackCallback = (deliveryTag, multiple) ->{
            //2.删除掉已经确认的消息 剩下的就是 未确认的消息
            // multiple为批量的情况 true
            if (multiple){
//                ConcurrentSkipListMap.head(K toKey, boolean inclusive);     小于或等于这个deliveryTag
//                ConcurrentSkipListMap.tailMap(K fromKey, boolean inclusive) 大于或等于这个deliveryTag
                //返回小于等于当前序列号的【已确认】的消息
                ConcurrentNavigableMap<Long, String> confirmed =
                        outStandingConfirms.headMap(deliveryTag, true);
                //消除该部分【已确认】的消息
                confirmed.clear();
            }else {
                //只消除当前序列号的消息
                outStandingConfirms.remove(deliveryTag);
            }
            System.out.println(">确认的消息：" + deliveryTag);
        };

        //消息确认失败 回调函数
        /**
         * deliveryTag:消息的标记
         * multiple:是否批量确认
         */
        ConfirmCallback nackCallback = (deliveryTag, multiple) ->{
            //3.打印一下未确认的消息有哪些，在成功的回调中已经删除了【已确认】的消息，就剩下未确认的了
            String message = outStandingConfirms.get(deliveryTag);
            System.out.println("未确认的消息：" + message);
        };

        //准备消息的监听器 监听哪些消息成功了 哪些消息失败了，如果不想要成功的回调可以设置为null
        channel.addConfirmListener(ackCallback, nackCallback); //异步操作

        long begin = System.currentTimeMillis();

        //批量发送
        for (int i = 0; i < message_count; i++) {
            String message = "消息" + i;
            channel.basicPublish("",queueName,null,message.getBytes("UTF-8"));
            //1.记录下所有要发送的消息的总和
            outStandingConfirms.put(channel.getNextPublishSeqNo(), message);
        }

        long end = System.currentTimeMillis();
        System.out.println("发布" +message_count+"个批量消息，耗时" + (end - begin) + "ms");
    }
}
